package service

import (
	"encoding/json"
	"fmt"
	"github.com/pkg/errors"
	"math"
	"strconv"
	"sync"
	"time"
	"zimuzu_cvn_web_api/config"
	"zimuzu_cvn_web_api/model"
	"zimuzu_cvn_web_api/pkg/drive/redis"
	"zimuzu_cvn_web_api/pkg/e"
	"zimuzu_cvn_web_api/pkg/errno"
)

// 监控日志
// Auth: hzp
func MonitorSMS(Title string, err error) {
	if err != nil {
		err = e.SendCronSMS("18827097756", Title, "失败")
	}
}

// 写日志
// Auth: hzp
func writerLog(i, success int) {
	e.Log("info", "【线程组", i, "】已完成 {", success, "} 条数据处理")
}

// 写进度
// Auth: hzp
func writerProcessLog(i, left, total int) {
	e.Log("info", "【线程组", i, "】进度为：", left, "/", total)
}

// 获取在线节点数
// Auth: hzp
// from [ 2 => Linux,  3 => Router, 1 => Pc ] 新的公式
func GetCurrentNode(From int) (nodeNum float64) {

	cacheKey := e.GlobalNodeNumForCache + strconv.Itoa(From)
	if redis.Exists(cacheKey) {
		rD, err := redis.Get(cacheKey)
		if err == nil {
			nodeNum, err := strconv.ParseFloat(rD, 64)
			if err == nil {
				return nodeNum
			}
		}
	}

	var nodeNumFInt int
	var err error
	if redis.Exists(e.StatisticalForTag) {
		nodeNumFInt, err = model.GetMgPaymentNodeNums(1, From)
		if err != nil {
			return e.PaymentFNodeNumVar
		}
	} else {
		// 从数据库取昨天的数据
		nodeNumFInt, err = model.GetMgPaymentNodeNums(0, From)
		if err != nil {
			return e.PaymentFNodeNumVar
		}
	}

	// 数据库取出的真实节点数
	nodeNumStr := strconv.Itoa(nodeNumFInt)
	nodeNumF64, err := strconv.ParseFloat(nodeNumStr, 64)

	// 基数
	var BaseNumFloat float64
	if From == 3 {
		// 路由器端
		BaseNumStr, err := model.GetGlobalVar(e.SettlementBaseRouterNodeNum)
		if err != nil {
			BaseNumFloat = e.SettlementBaseNode
		} else {
			BaseNumFloat, err = strconv.ParseFloat(BaseNumStr, 64)
			if err != nil {
				BaseNumFloat = e.SettlementBaseNode
			}
		}
	} else {
		// 非路由器端
		BaseNumStr, err := model.GetGlobalVar(e.SettlementBaseNodeNum)
		if err != nil {
			BaseNumFloat = e.SettlementBaseRouterNode
		} else {
			BaseNumFloat, err = strconv.ParseFloat(BaseNumStr, 64)
			if err != nil {
				BaseNumFloat = e.SettlementBaseRouterNode
			}
		}
	}

	nodeNum = nodeNumF64 + BaseNumFloat

	// 更新数据库数据用于前端显示
	var name string
	if From == 3 {
		name = e.SettlementRouterNum
	} else {
		name = e.SettlementNodeNum
	}
	err = model.UpdateGlobalVar(name, e.ParseString(nodeNum, 2))
	if err != nil {
		return nodeNum
	}
	redis.SetUnRe(cacheKey, nodeNum, e.TimeFormatMin*10)
	return nodeNum
}

// 更新全网收益能力数据
type CapacityData struct {
	GlobalCapacity     float64 `json:"globalCapacity"`
	LastGlobalCapacity float64 `json:"lastGlobalCapacity"`
}

func UpdateGlobalCapacity() (res CapacityData, err error) {

	cacheKey := e.PublicCacheKeyPre + "global:capacity"
	if redis.Exists(cacheKey) {
		redisStr, _ := redis.Get(cacheKey)
		err = json.Unmarshal([]byte(redisStr), &res)
		if err == nil {
			return res, nil
		}
		e.Log("info", err.Error())
	}

	res.GlobalCapacity, res.LastGlobalCapacity, err = model.GetGlobalCapacity()
	if err != nil {
		return res, err
	}

	if res.LastGlobalCapacity <= 0 {
		res.LastGlobalCapacity = res.GlobalCapacity
	}

	rData, _ := json.Marshal(res)
	redis.SetUnRe(cacheKey, string(rData), e.TimeFormatMin*5)

	return res, nil
}

// 统计指定用户的挖矿能力
// Auth: hzp
func doSuperMinerMiningCapacity(doStr string, doLeft, doTotalNum, doIk int) (err error) {

	// 写进度
	writerProcessLog(doIk, doLeft, doTotalNum)

	// 解析数据
	var uid int64
	err = json.Unmarshal([]byte(doStr), &uid)
	if err != nil {
		return err
	}

	// 更新挖矿能力
	res, err := UpdateGlobalCapacity()
	if err != nil {
		return err
	}

	err = model.GetSuperMinerMiningCapacity(uid, res.LastGlobalCapacity)
	if err != nil {
		return err
	}

	return
}

// 多线程处理方法
// Auth: hzp
func MultiThread(cacheKey string, totalNum, AgainNum int, do func(doStr string, doLeft, doTotalNum, doIk int) (doErr error)) (err error) {

	// 多开线程组
	var wg sync.WaitGroup
	if config.C.RunMode != "debug" {
		AgainNum = 20
	}
	wg.Add(AgainNum)
	for ik := 1; ik <= AgainNum; ik++ {
		var success int
		go func(cacheKey string, totalNum, ss int) {
			defer wg.Done()
			var left int
			for ii := 0; ii < totalNum; ii++ {
				left, _ = redis.LLen(cacheKey)
				if left <= 0 {
					break
				}
				tempStr, _ := redis.LPop(cacheKey)
				err = do(tempStr, left, totalNum, ik)
				if err == nil {
					success++
				} else {
					e.Log("err", err)
				}
			}
			writerLog(ss, success)
		}(cacheKey, totalNum, ik)
	}
	wg.Wait()
	e.Log("info", "线程组已全部完成")

	return
}

// 每天结算统计完再 计算租用令牌用户的收益能力
// Auth: hzp
func GetSuperMinerMiningCapacity() (err error) {

	e.Log("info", "统计矿工收益能力 开始执行")

	// 获取所以有矿工用户列表
	uidList, err := model.GetSuperMinerList()
	if err != nil {
		return errors.New("获取所以有质押是用户列表 错误：" + err.Error())
	}

	// 先写入队列供多线程
	var totalNum int
	cacheKey := e.TempSuperMinerList
	if redis.Exists(cacheKey) {
		return errors.New("队列没有执行完成，不能重复执行")
	}
	if len(uidList) <= 0 {
		return errors.New("队列为空，不需要处理")
	}
	for _, k := range uidList {
		kk, err := json.Marshal(k.Uid)
		if err == nil {
			_, err = redis.LPush(cacheKey, kk)
		}
		if err == nil {
			totalNum++
		}
	}

	// 多线程处理用户的初始化数据
	AgainNum := 10
	err = MultiThread(cacheKey, totalNum, AgainNum, doSuperMinerMiningCapacity)
	if err != nil {
		return err
	}

	return
}

// 网心结点流量数据获取
func GetMgNodeFlows() error {

	e.Log("info", "开始拉取流量")

	// 配置参数
	PageSize := 10000
	var StartTime, EndTime string
	// 隔天零点取第二天23点点数据
	if time.Now().Hour() == 0 {
		StartTime = time.Now().AddDate(0, 0, -1).Format("2006-01-02 22:00:00")
		EndTime = time.Now().AddDate(0, 0, -1).Format("2006-01-02") + " 23:59:59"
	} else {
		StartTime = time.Now().Format("2006-01-02 ") + fmt.Sprintf("%02d", time.Now().Hour()-2) + ":00:00"
		//StartTime = time.Now().Format("2006-01-02") + " 00:00:00"
		EndTime = time.Now().Format("2006-01-02") + " 23:59:59"
	}

	// 先请求一次拿去到要分页的数据
	page := 1
	res, err := GetYxFlowData(StartTime, EndTime, page, PageSize)
	if err != nil {
		return err
	}
	if res.Code != 0 {
		return err
	}
	Result := res.Result
	repeatNum := Result.Pagination.Total / int64(PageSize)
	listCacheKey := e.WxFlowsForMgCache
	if !redis.Exists(listCacheKey) {
		var i int64
		for i = 2; i <= repeatNum+1; i++ {
			_, _ = redis.LPush(listCacheKey, i)
		}
	}
	// 处理第一页的数据
	doneUpdateNodeFlow(page, res)
	e.Log("info", "开始处理第二页数据")

	// 多线程处理后面的数据
	repeatNumInt, _ := strconv.Atoi(strconv.FormatInt(repeatNum, 10))
	var wg sync.WaitGroup
	AgainNum := 1
	wg.Add(AgainNum)
	for ik := 1; ik <= AgainNum; ik++ {
		//var success int
		go func(cacheKey string, totalNum, ss int) {
			defer wg.Done()
			for {
				var left int
				left, _ = redis.LLen(cacheKey)
				if left <= 0 {
					break
				}
				tempStr, _ := redis.LPop(cacheKey)
				page, _ := strconv.Atoi(tempStr)
				res, err := GetYxFlowData(StartTime, EndTime, page, PageSize)
				if err != nil {
					continue
				}
				if res.Code != 0 {
					continue
				}
				doneUpdateNodeFlow(page, res)
				writerLog(ss, len(res.Result.FlowData))
			}
		}(listCacheKey, repeatNumInt, ik)
	}
	wg.Wait()
	e.Log("info", "线程组已全部完成")
	e.Log("info", "流量拉取完毕")

	return nil
}

// 网心结点数据获取
func GetMgNodeInfo() error {

	PageSize := 10000
	var StartTime, EndTime string
	StartTime = time.Now().Format("2006-01-02 00:00:00")
	EndTime = time.Now().Format("2006-01-02") + " 23:59:59"

	// 先请求一次拿去到要分页的数据
	page := 1
	res, err := GetNodeInfoData(StartTime, EndTime, page, PageSize)
	if err != nil {
		return err
	}
	if res.Code != 0 {
		e.Log("info", "请求接口失败：", res.Code)
		return errors.New("请求接口失败")
	}
	Result := res.Result
	repeatNum := Result.Pagination.Total / int64(PageSize)
	listCacheKey := e.WxNodeInfoForMgCache
	if !redis.Exists(listCacheKey) {
		var i int64
		for i = 2; i <= repeatNum+1; i++ {
			_, _ = redis.LPush(listCacheKey, i)
		}
	}
	// 处理第一页的数据
	doneUpdateNodeInfo(page, res)

	// 多线程处理后面的数据
	repeatNumInt, _ := strconv.Atoi(strconv.FormatInt(repeatNum, 10))
	var wg sync.WaitGroup
	AgainNum := 1
	wg.Add(AgainNum)
	for ik := 1; ik <= AgainNum; ik++ {
		//var success int
		go func(cacheKey string, totalNum, ss int) {
			defer wg.Done()
			for {
				var left int
				left, _ = redis.LLen(cacheKey)
				if left <= 0 {
					break
				}
				tempStr, _ := redis.LPop(cacheKey)
				page, _ := strconv.Atoi(tempStr)
				res, err := GetNodeInfoData(StartTime, EndTime, page, PageSize)
				if err != nil {
					continue
				}
				if res.Code != 0 {
					continue
				}
				doneUpdateNodeInfo(ss, res)
				writerLog(ss, len(res.Result.NodeData))
			}
		}(listCacheKey, repeatNumInt, ik)
	}
	wg.Wait()
	e.Log("info", "线程组已全部完成")

	return nil
}

// 更新节点信息
func doneUpdateNodeInfo(taskNum int, res model.YXInfoResponse) {

	if taskNum <= 0 {
		return
	}

	InfoData := res.Result.NodeData
	doNum := 0
	for _, k := range InfoData {
		UpdateMgNodeInfo(k.Node, k.Ip, k.Status, k.OnlineTime, k.DiskAvail)
		doNum++
	}
	e.Log("info", "【第", taskNum, "页{", doNum, "}条数据】已经处理完成")
	return
}

// 更新结点信息
func UpdateMgNodeInfo(NodeStr, Ip, Status string, OnlineTime int, DiskAvail float64) {

	NodeInfo, Way, err := GetMgNodeInfoByCache(NodeStr)
	if err != nil {
		e.Log("err", err)
		return
	}
	if NodeInfo.NodeId <= 0 {
		return
	}

	var StatusInt int
	if Status == "online" {
		StatusInt = 1
	} else {
		StatusInt = 2
	}

	// 在线时长更新
	HistoryOnlineTime := UpdateOnlineTimeDB(NodeInfo.HistoryOnlineTime, OnlineTime)
	isp, err := GetISPByIpByCache(Ip)
	if err != nil {
		e.Log("err", err)
		return
	}
	err = model.UpdateMgNodeInfo(NodeInfo.NodeId, NodeInfo.Sn, NodeInfo.Mac, Ip, HistoryOnlineTime, StatusInt, Way, OnlineTime, DiskAvail, isp)
	if err != nil {
		e.Log("err", err)
		return
	}
	return
}

//在线时长JSON数据更新
//Auth: Xu
func UpdateOnlineTimeDB(onlineTimeJson string, valtime int) string {
	onlineTimeMap := make(map[string]int, 0)
	json.Unmarshal([]byte(onlineTimeJson), &onlineTimeMap)
	var currentTimeStr string
	currentTime := time.Now()
	currentTimeStr = currentTime.Format("2006-01-02")
	if _, ok := onlineTimeMap[currentTimeStr]; len(onlineTimeMap) >= 7 && !ok {
		oldTime := currentTime.AddDate(0, 0, -7)
		oldTimeStr := oldTime.Format("2006-01-02")
		nextDayTime, _ := time.ParseInLocation("2006-01-02 15:04:05", oldTimeStr+" 00:00:00", time.Local)
		for k, _ := range onlineTimeMap {
			DayTimeStr, _ := time.ParseInLocation("2006-01-02 15:04:05", k+" 00:00:00", time.Local)
			if DayTimeStr.Unix() < nextDayTime.Unix() {
				delete(onlineTimeMap, k)
			}
		}
	}
	var hour float64
	if time.Now().Hour() == 23 {
		hour = (float64(valtime)*60)/3600 + 0.25
	} else {
		hour = (float64(valtime)*60)/3600 + 0.08
	}
	hourInt := math.Floor(math.Floor(hour))
	onlineTimeMap[currentTimeStr] = int(hourInt)
	timeJson, _ := json.Marshal(onlineTimeMap)

	return string(timeJson)
}

// 运营商数据获取
func GetISPByIpByCache(ip string) (res model.NodeSpModel, err error) {
	cacheKey := e.WxNodeIspForCache + ip
	if redis.Exists(cacheKey) {
		redisData, err := redis.Get(cacheKey)
		if err != nil {
			return res, err
		}
		err = json.Unmarshal([]byte(redisData), &res)
		if err != nil {
			return res, err
		}
		return res, nil
	}
	spInfo, err := model.GetISPByIp(ip)
	if err != nil {
		return spInfo, err
	}

	reData, err := json.Marshal(spInfo)
	if err != nil {
		return spInfo, err
	}
	redis.SetUnRe(cacheKey, string(reData), e.TimeFormatDay)
	return spInfo, nil
}

// 每个月给第三批用户发放CVN补贴
// Auth: hzp
func SubsidyForThirdRouter() (err error) {

	// 只补发一年的时间，到2020-08-29截至发放
	if time.Now().Year() >= 2020 {
		if time.Now().Year() == 2020 && time.Now().Month() > 8 {
			return
		}
	}

	// 统计数据
	err = model.AddNodePaymentForSub(0)
	if err != nil {
		e.Log("err", err.Error())
		return err
	}

	cacheKey := e.SubsidyRouterCache
	// 然后将数据加入到 redis List
	e.Log("info", "【start】将需要更新的数据添加到Redis队列")
	redis.UnDelete(cacheKey)
	if !redis.Exists(cacheKey) {
		UidList, err := model.GetThirdRouterUserList()
		if err == nil && len(UidList) > 0 {
			for i := 0; i < len(UidList); i++ {
				redis.UnLPush(cacheKey, UidList[i].Uid)
			}
		}
	}
	e.Log("info", "【end】")

	for {
		e.Log("info", "开始处理")
		Len, err := redis.LLen(cacheKey)
		if err != nil && Len <= 0 {
			return nil
		}

		UidStr, err := redis.LPop(cacheKey)
		if err != nil {
			e.Log("info", "用户补发完成")
			return nil
		}
		Uid, _ := strconv.ParseInt(UidStr, 10, 64)

		// 获取到需要发放的 CVNT 数据
		k, err := model.GetThirdRouterPaymentByUid(Uid)
		if err != nil {
			return errno.RecordError
		}

		err = thirdRouterUpdateStatus(k.Id, Uid, k.CVNT)
		if err != nil {
			return err
		}

		// 发放完成之后清理页面缓存
		ReloadLinuxNodePaymentByUid(Uid)
	}
}

// 更新第三批用户发放数据
// Auth: hzp
// 1、更新用户账目表 cvn_log
// 2、请求总账目接口 user_amount
// 3、更新流水表
// 4、更新 状态
func thirdRouterUpdateStatus(LogId, Uid int64, Cvn float64) (err error) {

	// 开启事务
	conn, err := config.C.DB.DB.Beginx()
	if err != nil {
		return err
	}

	// 1、修改amount,插入cvn流水
	updateAmountData := UpdateCvnAmountParams{
		Uid:         Uid,
		Amount:      Cvn,
		Source:      model.LogSourceRouterAward,
		RelateId:    strconv.FormatInt(LogId, 10),
		RelateTable: "mg_user_payment",
	}
	_, err = UpdateCvnAmountBalance(conn, updateAmountData)
	if err != nil {
		conn.Rollback()
		return err
	}

	// 2、更新状态
	err = model.UpdateUserPayment(conn, LogId)
	if err != nil {
		conn.Rollback()
		return err
	}

	conn.Commit()
	return nil
}

// 释放缓存
func ReloadLinuxNodePaymentByUid(Uid int64) {
	DataStr := time.Now().AddDate(0, 0, -1).Format("2006-01-02")
	cacheKey := e.PublicCacheKeyPre + "linux:node:payment:" + strconv.FormatInt(Uid, 10) + DataStr
	redis.UnDelete(cacheKey)
}

// 保证金不足，每天12点自动从余额补充到保证金
// Auth: hzp
func SupplyTokenPledgingCvn() error {

	e.Log("info", "开始进行 SupplyTokenPledgingCvn")

	// 获取到需要缴纳保证金的用户
	uidList, err := model.GetNeedTokenPledgingCvn()
	if err != nil {
		return err
	}
	e.Log("info", "需要处理", len(uidList), "条数据")

	// 更新保证金
	fail := 0
	for _, k := range uidList {
		e.Log("info", "正在处理 uid=", k.Uid)
		err = UpdateTokenPledgingCvn(k.Uid)
		if err != nil {
			fail++
			e.Log("err", "更新保证金失败：", "uid=", k.Uid, " err: ", err.Error())
		}
		if err == errno.PledgedUnEnoughErr {
			// 直接释放
			err = ReleaseToken(k.Uid)
			if err != nil {
				e.Log("err", "释放保证金失败，", err.Error())
			}
		}
		e.Log("info", "uid= ", k.Uid, "处理完成")
	}

	if fail > 0 {
		errInfo := "自动补充保证金失败，失败" + strconv.Itoa(fail) + "个，总数" + strconv.Itoa(len(uidList)) + "个"
		return errors.New(errInfo)
	}

	e.Log("info", len(uidList), "条数据已全部处理完成")
	e.Log("info", "结束 SupplyTokenPledgingCvn")
	return nil
}

// 更新 应缴纳保证金
// Auth: hzp
func UpdateTokenPledgingCvn(Uid int64) (err error) {

	res, err := CheckRentTokenPledgeStatus(Uid, 0)
	if err != nil {
		return err
	}

	if res.Status == 1 {
		e.Log("info", "保证金充足，不需要缴纳")
		return nil
	}

	e.Log("params", res)

	// 不需要更新
	if res.Cvn <= 0 {
		return nil
	}

	// 开启事务
	conn, err := config.C.DB.DB.Beginx()
	if err != nil {
		return err
	}

	e.Log("info", res)

	// 1、修改amount,插入cvn流水
	updateAmountData := UpdateCvnAmountParams{
		Uid:          Uid,
		Amount:       -res.NeedCvn,
		FrozenAmount: res.NeedCvn,
		Source:       model.LogSourceSystemAddDeposit,
	}

	e.Log("info", updateAmountData)
	_, err = UpdateCvnAmountBalance(conn, updateAmountData)
	if err != nil {
		conn.Rollback()
		return err
	}

	// 2、增加保证金
	err = UpdateCvnFrozenAmount(conn, Uid, res.NeedCvn, 0, model.PledgeDeposit)
	if err != nil {
		conn.Rollback()
		return err
	}

	conn.Commit()
	e.Log("info", "用户Uid {", Uid, "}应缴保证金 {", res.NeedCvn, " }CVNT 已完成")

	return nil
}

// 释放令牌
// Auth: hzp
func ReleaseToken(uid int64) (err error) {

	// 找到该用户租用的令牌订单
	orderList, err := model.GetUserStartingOrder(uid)
	if err != nil {
		return err
	}

	tokenList, err := model.GetMgTokenListByUserId(uid)
	if err != nil {
		return err
	}

	// 逐个订单释放
	fail := 0
	for _, k := range orderList {
		err = model.UpdateMgTokenAndLog(k, 5, uid)
		if err != nil {
			fail++
		}
	}

	for _, kk := range tokenList {
		err = model.UnbindMgTokenAndLog(kk, 5)
		if err != nil {
			fail++
		}
	}

	if fail > 0 {
		errInfo := "释放令牌失败，失败" + strconv.Itoa(fail) + "个，总数" + strconv.Itoa(len(orderList)) + "个"
		return errors.New(errInfo)
	}

	return
}

// 新版结算 统计
// @Time: 每天凌晨00:10开始统计
// **一个用户一个线程
// 1、统计出要结算的用户、写入队列
// 2、队列取出一个用户，提前准备以下信息
//     2.1 用户的质押信息
//     2.2 用户的结点数据信息、包含在线时长
//     2.3 用户的超级结点数据信息
//     2.4 用户对应结点的流量信息
//     2.5 用户对应结点的结算收益信息
// 3、处理用户的结算信息、事务处理
// 4、插入到待发放记录表
func StatisticalStageForYx() (err error) {

	e.Log("info", "开始进入统计程序")
	TagKey := e.StatisticalForTag
	if redis.Exists(TagKey) {
		return errno.StatisticalAgainErr
	}
	redis.SetUnRe(TagKey, 1, -1)
	e.Log("info", "记录程序运行标示")

	// TODO  将需要结算的用户写入到redis队列
	var totalNum int
	cacheKey := e.StatisticalForUser
	redis.UnDelete(cacheKey)
	if !redis.Exists(cacheKey) {
		NodeList, err := model.GetMgNodeInfoForRedis()
		if err != nil {
			return errno.AgainStaErr
		}
		for _, k := range NodeList {
			kk, err := json.Marshal(k)
			if err == nil {
				_, err = redis.LPush(cacheKey, kk)
			}
			if err == nil {
				totalNum++
			}
		}
	}

	// TODO 多线程处理
	var wg sync.WaitGroup
	AgainNum := 100
	if config.C.RunMode != "debug" {
		AgainNum = 100
	}
	wg.Add(AgainNum)
	for ik := 1; ik <= AgainNum; ik++ {
		var success int
		go func(cacheKey string, totalNum, ss int) {
			defer wg.Done()
			var left int
			for ii := 0; ii < totalNum; ii++ {
				left, _ = redis.LLen(cacheKey)
				if left <= 0 {
					break
				}
				tempStr, _ := redis.LPop(cacheKey)
				err = writeYxNodeData(tempStr, left, totalNum, ss)
				if err == nil {
					success++
				} else {
					e.Log("err", err)
				}
			}
			writerLog(ss, success)
		}(cacheKey, totalNum, ik)
	}
	wg.Wait()
	e.Log("info", "线程组已全部完成")

	err = model.AddMgUserPayment()
	if err != nil {
		return err
	}

	e.Log("info", "清理程序运行标示")
	redis.UnDelete(TagKey)
	e.Log("info", "统计程序已全部完成")

	return nil
}

// 写记录到node_payment表
func writeYxNodeData(dateStr string, left, totalNum, ik int) (err error) {

	// 写进度
	writerProcessLog(ik, left, totalNum)

	// 解析redis数据
	var NodeInfo model.MgNodeInfoResponse
	err = json.Unmarshal([]byte(dateStr), &NodeInfo)
	if err != nil {
		return err
	}

	if NodeInfo.NodeId <= 0 || NodeInfo.Uid <= 0 {
		e.Log("info", "[1] NodeId=", NodeInfo.NodeId, " Uid=", NodeInfo.Uid, errno.IgnoreTask.Info)
		return errno.IgnoreTask
	}

	// 数据校验
	res, err := checkNodeInfo(NodeInfo, 1)
	if err != nil {
		return errno.IgnoreTask
	}

	// 事物处理将数据写入到node_payment数据表
	err = model.AddMgNodePaymentRecord(res)
	if err != nil {
		e.Log("info", "[7] NodeId=", NodeInfo.NodeId, " Uid=", NodeInfo.Uid, errno.IgnoreTask.Info)
		return errno.IgnoreTask
	}

	return nil
}

// 结算和业务APP展示预估收益
// 字段说明：Show 1 需要考虑在线时长  0 不需要考虑在线时长
// Auth: hzp
func checkNodeInfo(NodeInfo model.MgNodeInfoResponse, Show int) (res model.CheckNodeInfoResponse, err error) {

	e.Log("info", "正在校验结点{nodeId=", NodeInfo.NodeId, "}数据")

	res.Uid = NodeInfo.Uid
	res.NodeId = NodeInfo.NodeId
	res.Type = NodeInfo.Type

	// 2.1 改节点是否有令牌
	res.PledgeStatus = 0
	if NodeInfo.TokenStatus == 1 {
		e.Log("info", "结点【", NodeInfo.Uid, " / ", NodeInfo.NodeId, "】结算时质押状态为: ", NodeInfo.TokenStatus)
	} else {
		res.PledgeStatus = 1
	}

	// 2.2 用户的结点数据信息、包含在线时长
	var DateStr string
	if Show == 1 {
		DateStr = time.Now().AddDate(0, 0, -1).Format("2006-01-02")
	} else {
		DateStr = time.Now().Format("2006-01-02")
	}
	mapData := make(map[string]int, 0)
	err = json.Unmarshal([]byte(NodeInfo.HistoryOnlineTime), &mapData)
	if err != nil {
		e.Log("info", "[3] NodeId=", NodeInfo.NodeId, " Uid=", NodeInfo.Uid, errno.IgnoreTask.Info)
		res.OnlineTime = 0
	} else {
		if _, ok := mapData[DateStr]; !ok {
			res.OnlineTime = 0
		} else {
			res.OnlineTime = mapData[DateStr]
		}
	}
	res.DateStr = DateStr

	// 2.3 用户的超级结点数据信息
	res.SuperNodeId, err = GetSuperNodeIdByUidFromCache(NodeInfo.Uid)
	if err != nil {
		e.Log("info", "[5] NodeId=", NodeInfo.NodeId, " Uid=", NodeInfo.Uid, errno.IgnoreTask.Info)
		res.SuperNodeId = 0
	}

	// 2.4 用户对应结点的流量信息
	res.Up, err = model.GetUnMgNodeInfoForRedis(NodeInfo.NodeId, Show)
	if err != nil {
		e.Log("info", "[6] NodeId=", NodeInfo.NodeId, " Uid=", NodeInfo.Uid, errno.IgnoreTask.Info)
		res.Up = 0.00
	}

	// 2.5 用户对应结点的结算收益信息
	res.IncomeMode, res.Cvn, err = GetMgNodeProfit(NodeInfo.Uid, res.SuperNodeId, res.Up, res.OnlineTime, Show, NodeInfo.Type, res.PledgeStatus, NodeInfo.Batch)
	if err != nil {
		return res, err
	}
	//e.Log("err", "[获取结点收益数据如下:] nodeId=", NodeInfo.NodeId, "&up=", res.Up, "&Cvn=", res.Cvn)

	return res, nil
}

// 缓存用户 超级结点信息
func GetSuperNodeIdByUidFromCache(Uid int64) (SuperNodeId int64, err error) {
	cacheKey := e.StatisticalForCache + ":superNodeId:" + strconv.FormatInt(Uid, 10)
	if redis.Exists(cacheKey) {
		reData, err := redis.Get(cacheKey)
		if err != nil {
			return SuperNodeId, err
		}
		SuperNodeId, _ = strconv.ParseInt(reData, 10, 64)
		return SuperNodeId, nil
	}

	SuperNodeId, err = model.GetSuperNodeMinerByUid(Uid)
	if err != nil {
		SuperNodeId = 0
	}
	redis.SetUnRe(cacheKey, SuperNodeId, e.TimeFormatMin*10)
	return
}

// 校验是否加入到了黑名单
func CheckUserIsBlack(uid int64) int {

	err := model.GetBlackUser(uid)
	if err != nil {
		return 0
	}
	return 1
}

// 获取单结点收益
func GetMgNodeProfit(Uid, SuperNodeId int64, flows float64, OnlineTime, Show, From, PledgeStatus, Batch int) (incomeModel int, cvn float64, err error) {

	//校验是否在黑名单
	if CheckUserIsBlack(Uid) == 1 {
		return 0, 0.00, errno.IgnoreTask
	}

	income := e.IncomeBase
	if Show == 1 {
		// 在线时长数据判断  20个小时
		switch From {
		case model.NodeTypeLinux:
			if OnlineTime < e.PaymentHourLimit {
				return 0, 0.00, errno.IgnoreTask
			}
			if PledgeStatus <= 0 {
				// 质押状态判断
				return 0, 0.00, errno.IgnoreTask
			}
			if SuperNodeId <= 0 {
				// 超级节点判断
				return 0, 0.00, errno.IgnoreTask
			}
			break
		case model.NodeTypeRouter:
			if OnlineTime < e.RouterPaymentHourLimit {
				return 0, 0.00, errno.IgnoreTask
			}
			if PledgeStatus <= 0 {
				// 质押状态判断
				return 0, 0.00, errno.IgnoreTask
			}
			if SuperNodeId <= 0 {
				// 超级节点判断
				return 0, 0.00, errno.IgnoreTask
			}
			break
		case model.NodeTypePc:
			if PledgeStatus == 1 && (OnlineTime < e.PcPaymentHourLimit || SuperNodeId <= 0) {
				income = 0.00
			} else if PledgeStatus == 0 {
				income = 0.00
			}
		}
	} else {
		if PledgeStatus <= 0 {
			if From == 2 || From == 3 {
				return 0, 0.00, errno.IgnoreTask
			}
			income = 0.00
		} else {
			if From == 1 && SuperNodeId <= 0 {
				income = 0.00
			}
		}
	}

	dynamicIncome, err := DynamicIncome(income, flows, From, Batch)
	if err != nil {
		e.Log("info", "校验收益失败", err)
		return 0, cvn, err
	}

	if Show == 0 {
		dynamicIncome = 0.9 * dynamicIncome
	}

	return 1, dynamicIncome, nil
}

// 动态收益
// from [ 0 => Linux,  1 => Router, 2 => Pc ] 旧的公式
// from [ 2 => Linux,  3 => Router, 1 => Pc ] 新的公式
func DynamicIncome(inCome, flows float64, From, Batch int) (cvn float64, err error) {

	//Log("info", "动态收益，inCome=", inCome, " &flows=", flows, " &From=", From, " &Batch=", Batch)

	// 路由器 动态收益f
	var DynamicIncome float64
	switch From {
	case 2:
		DynamicIncome = math.Floor(flows*1024) * e.BaseDynamicFlows / 100
		break
	case 3:
		DynamicIncome = math.Floor(flows) * e.RouterDynamicFlows
		break
	case 1:
		DynamicIncome = flows * e.RouterDynamicFlows
		break
	default:
		DynamicIncome = math.Floor(flows) * e.RouterDynamicFlows
		break
	}

	// 每日贡献流量
	// 低于10M的，结算0%的代币；
	// 高于10M低于20M的，结算10%的代币；
	// 高于20M低于30M的，结算20%的代币；
	// 高于30M低于40M的，发放30%的代币；
	// 高于40M低于50M的，发放40%的代币；
	// 高于100M的结算100%的代币
	BaseNum, err := strconv.ParseFloat(e.ParseString(flows*102.400, 0), 0)
	if err != nil {
		return cvn, err
	}

	if BaseNum >= 9.999 {
		BaseNum = 1.00
	} else if BaseNum < 1.00 {
		BaseNum = 0.00
	} else {
		BaseNum = 0.1 * BaseNum
	}
	if From == 3 {
		// 新规则路由器流量小于10M全部按照10%收益给
		if flows <= 0 {
			BaseNum = 1.00 // 后续流量起来了会调整为 0.10
		} else {
			BaseNum = 1.00
		}
	}

	// 实行减产 Time:2019-09-17
	// n := math.Ceil(PaymentFConstVar * math.Pow(PaymentFConstFirstCvn/ cvnPrice, PaymentFConstPowVar))
	var minF float64
	if Batch != 4 {
		nodeNum := GetCurrentNode(From)
		f := e.PaymentFConstFVar * (math.Pow(nodeNum, e.PaymentFConstFPowVar))
		minF = math.Min(f, 1.000)
		//Log("info", "实行减产: modeNum=", nodeNum, " &f=", f, " &minF=", minF)
	} else {
		// 第四批路由器收益 f = 1
		minF = 1.00
	}

	//Log("info", "结算收益，DynamicIncome=", DynamicIncome, " &inCome=", inCome, " &minF=", minF, " &BaseNum=", BaseNum)
	cvn = (DynamicIncome + inCome) * minF * BaseNum

	//Log("info", "动态收益，cvn=", cvn)

	return cvn, nil
}

// 租赁用户的结算统计
// Auth: hzp
func PaymentForRentUser() error {

	return model.GetTokenPaymentNeedData()
}

// 新版结算 发放
// @Time: 每天上午09:59开始发放
// **一个用户一个线程
// 1、统计出要发放的用户、写入队列
// 2、逐个用户发放
func PaymentForMgNode() (err error) {

	cacheKey := e.PaymentForMgCache
	// 然后将数据加入到 redis List
	e.Log("info", "【2.0】【start】将需要更新的数据添加到Redis队列")
	var totalNum int
	if !redis.Exists(cacheKey) {
		UidList, err := model.GetMgPurchaseUserList()
		totalNum = len(UidList)
		if err == nil && totalNum > 0 {
			for i := 0; i < len(UidList); i++ {
				redis.UnLPush(cacheKey, UidList[i].Uid)
			}
		}
	}
	e.Log("info", "【2.0】【end】")

	// TODO 多线程处理
	var wg sync.WaitGroup
	AgainNum := 10
	if config.C.RunMode != "debug" {
		AgainNum = 10
	}
	wg.Add(AgainNum)
	for ik := 1; ik <= AgainNum; ik++ {
		var success int
		go func(cacheKey string, totalNum, ss int) {
			defer wg.Done()
			var left int
			for ii := 0; ii < totalNum; ii++ {
				left, _ = redis.LLen(cacheKey)
				if left <= 0 {
					break
				}
				UidStr, err := redis.LPop(cacheKey)
				if err != nil {
					e.Log("info", "用户发放完成")
					break
				}
				Uid, _ := strconv.ParseInt(UidStr, 10, 64)
				err = paymentByUid(Uid, left, totalNum, ss)
				if err == nil {
					success++
				} else {
					e.Log("err", err)
				}
			}
			writerLog(ss, success)
		}(cacheKey, totalNum, ik)
	}
	wg.Wait()
	e.Log("info", "线程组已全部完成")

	return
}

// 逐一发放CVNT
// Auth: hzp
func paymentByUid(Uid int64, left, totalNum, ik int) (err error) {

	// 写进度
	writerProcessLog(ik, left, totalNum)

	// 获取到需要发放的 CVNT 数据
	CvnDetail, err := model.GetMgUserPaymentByUid(Uid)
	if err != nil {
		return errno.RecordError
	}
	if len(CvnDetail) <= 0 {
		return nil
	}
	for _, k := range CvnDetail {
		if k.Id <= 0 {
			break
		}
		err = updateUserPaymentStatus(k.Id, Uid, k.Type, k.CVNT)
		if err != nil {
			e.Log("err", "uid= ", Uid, " 发放异常 :", err.Error())
		}
	}

	// 发放完成之后清理页面缓存
	ReloadLinuxNodePaymentByUid(Uid)
	return nil
}

// 单个用户发放CVNT
// Auth: hzp
// 1、更新用户账目表
// 2、请求总账目接口
// 3、更新流水表  cvn_log、user_amount
// 4、更新 状态   mg_user_payment
func updateUserPaymentStatus(LogId, Uid int64, Type int, Cvn float64) (err error) {

	// 开启事务
	conn, err := config.C.DB.DB.Beginx()
	if err != nil {
		return err
	}

	// 判断是否为令牌 结算收益
	tag := 0          // 标识：是否需要扣除保证金 0-否 1-全部扣除保证金 2-部分扣除保证金
	Source := 0       // 来源
	updateCvn := 0.00 // 额外补充的保证金数
	switch Type {
	case 1:
		Source = model.LogSourcePC
		break
	case 2:
		Source = model.LogSourceLX
		break
	case 3:
		Source = model.LogSourceRouter
		break
	case 4:
		Source = model.LogSourceRouterAward
		break
	case 5:
		Source = model.LogSourceTokenLend
		break
	case 6:
		Source = model.LogSourceTokenRent // 令牌租用扣款
		// 判断余额
		cvnAmount, err := GetUserCvnAmountByUid(Uid)
		if err != nil {
			return err
		}
		// 查询保证金字段
		userFrozen, err := model.GetCvnFrozenInfo(Uid, model.PledgeDeposit)
		if err != nil {
			return err
		}
		// 判断余额是否足够扣款
		if cvnAmount.Balance+Cvn >= 0 {
			// 只需要扣除可用余额
			Source = model.LogSourceTokenRent // 令牌租用扣款
		} else if cvnAmount.Balance <= 0 {
			if userFrozen.Amount+Cvn < 0 {
				return errno.PledgedUnEnoughErr
			}
			updateCvn = Cvn
			// 余额为0时，需要先将保证金转移到可用余额，再扣除可用余额
			tag = 1
		} else {
			if userFrozen.Amount+Cvn+cvnAmount.Balance < 0 {
				return errno.PledgedUnEnoughErr
			}
			updateCvn = Cvn + cvnAmount.Balance
			// 余额不足时，需要先将保证金转移到可用余额，再扣除可用余额
			tag = 2
		}
		break
	default:
		return errno.AgainErr
	}

	// 是否需要扣除保证金
	if tag > 0 {
		// 转出保证金
		err = UpdateCvnFrozenAmount(conn, Uid, updateCvn, 0, model.PledgeDeposit)
		if err != nil {
			return err
		}
	}

	// 3、修改amount,插入cvn流水,
	// 扣除余额
	updateAmountData := UpdateCvnAmountParams{
		Uid:         Uid,
		Amount:      Cvn,
		Source:      Source,
		RelateId:    strconv.FormatInt(LogId, 10),
		RelateTable: "mg_user_payment",
	}
	_, err = UpdateCvnAmountBalance(conn, updateAmountData)
	if err != nil {
		conn.Rollback()
		return err
	}

	// 4、更新 状态   mg_user_payment
	err = model.UpdateMgUserPaymentData(conn, LogId)
	if err != nil {
		conn.Rollback()
		return err
	}

	conn.Commit()
	e.Log("info", "用户Uid {", Uid, "& type= ", Type, "} 发放完成")
	return nil
}

// 发放 超级结点
func GrantForSuperNode() (err error) {

	// 只补发一年的时间，到2020-08-17 截至发放
	if time.Now().Year() >= 2019 {
		if time.Now().Year() == 2019 && time.Now().Month() >= 9 && time.Now().Day() > 18 {
			e.Log("info", "已经超时，不再处理该脚本了~")
			return
		}
	}

	UidList, err := model.GetSuperUserList()
	for _, k := range UidList {
		SuperCvn, err := model.GetSuperPaymentBySuperNodeId(k.NodeId)
		if err != nil {
			return errno.RecordError
		}

		err = superPaymentUpdateStatus(k.Uid, k.NodeId, SuperCvn)
		if err != nil {
			e.Log("err", "【UID = ", k.Uid, "】处理失败")
			return err
		}
		e.Log("info", "【UID = ", k.Uid, "】处理成功")
	}

	e.Log("info", "超级结点发放完成")
	return nil
}

func superPaymentUpdateStatus(Uid, NodeId int64, Cvn float64) (err error) {

	// 循环用户
	// 1、更新用户账目表 cvn_log
	// 2、请求总账目接口 user_amount
	// 3、更新 super_node_payment 表

	// 开启事务
	conn, err := config.C.DB.DB.Beginx()
	if err != nil {
		return err
	}

	// 1、修改amount,插入cvn流水,
	// 扣除余额
	updateAmountData := UpdateCvnAmountParams{
		Uid:    Uid,
		Amount: Cvn,
		Source: model.LogSourceSuperNodeMinerIncome,
	}
	_, err = UpdateCvnAmountBalance(conn, updateAmountData)
	if err != nil {
		conn.Rollback()
		return err
	}

	// 更新 super_node_payment 表
	err = model.InsertSuperNodePayment(conn, NodeId, Cvn)
	if err != nil {
		conn.Rollback()
		return err
	}
	conn.Commit()

	return nil
}
